package com.superbytecode.cloud.rocketmq.producer.provider;

import com.superbytecode.cloud.common.context.SpringContextUtil;
import com.superbytecode.cloud.common.model.mq.DelayedLevel;
import com.superbytecode.cloud.common.model.mq.MQConfigure;
import com.superbytecode.cloud.common.utils.AssertUtil;
import com.superbytecode.cloud.common.utils.VerifyUtil;
import com.superbytecode.cloud.rocketmq.exception.MQBizException;
import com.superbytecode.cloud.rocketmq.producer.MQProduceProvider;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Objects;

/**
 * All rights Reserved, Designed By www.super-master.com
 *
 * @ProjectName: beacon-cloud
 * @Package: com.superbytecode.cloud.rocketmq.producer
 * @ClassName: OpenMQProduceProvider
 * @Description: [开源RocketMq生产信息实现]
 * <p> </p>
 * @Author: [Lance Ting]
 * @Date: 2024/11/1 16:31
 * @Version: V1.0
 * @Copyright: 2024 www.super-master.com Inc. All rights reserved.
 * TODO: 注意,本文件Lance Ting所作,如果转载或使用请标明具体出处!
 **/
@Slf4j
@Component
public class OpenMQProduceProvider extends RocketMQTemplate implements MQProduceProvider {
    private final RocketMQTemplate rocketTemplate = SpringContextUtil.getBean("rocketMQTemplate", RocketMQTemplate.class);

    /**
     * 开源版本mq生产者
     *
     * @param topic     消息主题
     * @param configure 消息参数配置
     * @param body      消息内容
     */
    @Override
    public void producer(String topic, MQConfigure configure, Object body) {
        AssertUtil.requireNonEmpty(body, "message body must not be empty");
        // 实例化生产者
        DefaultMQProducer producer = rocketTemplate.getProducer();
        Message rocketMessage = this.obtainOpenMQMessage(topic, configure, body);
        SendResult result;
        try {
            String hash = configure.getHash();
            int sendMsgTimeout = producer.getSendMsgTimeout();
            if (VerifyUtil.isEmpty(hash)) {
                result = producer.send(rocketMessage, sendMsgTimeout);
            } else {
                MessageQueueSelector selector = rocketTemplate.getMessageQueueSelector();
                result = producer.send(rocketMessage, selector, hash, sendMsgTimeout);
            }
            log.info("openMQ send result topic [{}] msgId [{}] ", topic, result.getMsgId());
        } catch (Exception e) {
            Thread.currentThread().interrupt();
            log.error("openMQ send message ", e);
            throw new MQBizException("mq发送消息失败", e);
        }
    }

    /**
     * 获取开源MQ消息对象
     *
     * @param topic     消息主题
     * @param configure 配置
     * @param body      消息内容
     * @return org.apache.rocketmq.common.message.Message
     */
    private Message obtainOpenMQMessage(String topic, MQConfigure configure, Object body) {
        // 创建Message实例
        HashMap<String, Object> headersMap = new HashMap<>();
        MessageHeaders messageHeaders = new MessageHeaders(headersMap);
        org.springframework.messaging.Message<Object> message = MessageBuilder.createMessage(body, messageHeaders);
        //
        MessageConverter messageConverter = rocketTemplate.getMessageConverter();
        setMessageConverter(messageConverter);
        org.springframework.messaging.Message<?> msg = super.doConvert(message.getPayload(), message.getHeaders(), null);
        //
        Message rocketMessage = RocketMQUtil.convertToRocketMessage(messageConverter, "UTF-8", topic, msg);
        //
        if (Objects.isNull(configure)) {
            return rocketMessage;
        }
        String tag = configure.getTag();
        if (VerifyUtil.isNotEmpty(tag)) {
            rocketMessage.setTags(tag);
        }
        DelayedLevel delayedLevel = configure.getDelayedLevel();
        if (VerifyUtil.isNotEmpty(delayedLevel)) {
            rocketMessage.setDelayTimeLevel(delayedLevel.obtainCode());
        }
        return rocketMessage;
    }
}